package com.bw.dwd;

import com.bw.bean.DeviceData;
import com.bw.bean.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.alibaba.fastjson.JSON;

/**
 * 实时数据源封装（匹配文档数据来源）
 * 文档参考：🔶1-27（行为日志）、🔶1-175（设备数据）、🔶1-171（会员资料）
 */
public class ProfileSource {
    // Kafka集群地址（需替换为实际地址）
    private static final String KAFKA_BOOTSTRAP_SERVERS = "kafka01:9092,kafka02:9092";
    // MySQL连接信息（需替换为实际配置）
    private static final String MYSQL_URL = "jdbc:mysql://mysql01:3306/user_db?useSSL=false";
    private static final String MYSQL_USER = "root";
    private static final String MYSQL_PWD = "123456";

    /**
     * 1. 接入Kafka用户行为流（主题：user_behavior_topic）
     */
    public static DataStream<UserBehavior> getUserBehaviorSource(StreamExecutionEnvironment env) {
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(KAFKA_BOOTSTRAP_SERVERS)
                .setTopics("user_behavior_topic")
                .setGroupId("user_behavior_group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        return env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "UserBehaviorKafkaSource")
                .map(jsonStr -> JSON.parseObject(jsonStr, UserBehavior.class))
                .name("UserBehaviorJsonParse");
    }

    /**
     * 2. 接入Kafka设备数据流（主题：device_data_topic）
     */
    public static DataStream<DeviceData> getDeviceDataSource(StreamExecutionEnvironment env) {
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(KAFKA_BOOTSTRAP_SERVERS)
                .setTopics("device_data_topic")
                .setGroupId("device_data_group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        return env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "DeviceDataKafkaSource")
                .map(jsonStr -> JSON.parseObject(jsonStr, DeviceData.class))
                .name("DeviceDataJsonParse");
    }

    /**
     * 3. 接入MySQL CDC用户基础信息流（表：user_base）
     */
    public static DataStream<String> getUserBaseCDCSource(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("mysql01")
                .port(3306)
                .databaseList("user_db") // 数据库名
                .tableList("user_db.user_base") // 表名
                .username(MYSQL_USER)
                .password(MYSQL_PWD)
                .deserializer(new JsonDebeziumDeserializationSchema()) // 输出JSON格式
                .build();

        return env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "UserBaseCDCSource");
    }
}